package cn.basttg.demo.zookeeper;

import java.io.UnsupportedEncodingException;
import java.util.List;

import net.sf.json.JSONObject;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class ZkWebmccWatcher extends Thread implements Watcher {
	private ZooKeeper zk = null;
	private Integer mutex;
	private String root;
	private boolean running = true; 

	synchronized public void process(WatchedEvent event) {
		synchronized (mutex) {
			// System.out.println("Process: " + event.getType());
			mutex.notify();
		}
	}

	@Override
	public void run() {
		while(running) {
			try {
				consume();
				System.out.println("ZkWebmccWatcher success!");
            } catch (KeeperException e){
            	System.out.println("ZkWebmccWatcher error: ");
            	e.printStackTrace();
            } catch (InterruptedException e){

            }
		}
	}
	
	public void stopRunning() {
		running = false;
	}

	/**
	 * Constructor of producer-consumer queue
	 * @param connectString
	 * @param rootPath
	 */
	ZkWebmccWatcher(String connectString, String rootPath) {
		if (zk == null) {
			try {
				System.out.println("Starting ZkWebmccWatcher:");
				zk = new ZooKeeper(connectString, 3000, this);
				mutex = new Integer(-1);
				System.out.println("Finished starting ZkWebmccWatcher: " + zk);
			} catch (Exception e) {
				System.out.println(e.toString());
				zk = null;
			}
		}
		// else mutex = new Integer(-1);

		this.root = rootPath;
		// Create ZK node name
		if (zk != null) {
			try {
				Stat s = zk.exists(root, false);
				if (s == null) {
					zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
				}
			} catch (KeeperException e) {
				System.out.println("Keeper exception when instantiating queue: " + e.toString());
			} catch (InterruptedException e) {
				System.out.println("Interrupted exception");
			}
		}
	}

	/**
	 * Remove first element from the queue.
	 * @return
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	void consume() throws KeeperException, InterruptedException {
		while (true) {
			synchronized (mutex) {
				List<String> list = zk.getChildren(root, true);
				if (list.size() == 0) {
					System.out.println("ZkWebmccWatcher Going to wait");
				} else {
					System.out.println("\t----" + root);
					try {
						for (String s : list) {
							byte[] b = zk.getData(root + "/" + s, false, null);
							String value = new String(b, "UTF-8");
							JSONObject meeting = JSONObject.fromObject(value);
							String meetingRoomId = meeting.getString("meetingRoomId");
							String shortId = meeting.getString("shortId");
							String subject = meeting.getString("subject");
							System.out.println("\t      └─" + meetingRoomId + ":  " + shortId + " ==> " + subject);
						}
					} catch (UnsupportedEncodingException e) {
						e.printStackTrace();
					}
				}
				mutex.wait();
			}
		}
	}

	public static void main(String args[]) {
		String connectionString = "172.16.160.196:2181";
		String rootPath = "/app1";

		ZkWebmccWatcher mcu = new ZkWebmccWatcher(connectionString, rootPath);
		try {
			mcu.consume();
			System.out.println("ZkWebmccWatcher success!");
        } catch (KeeperException e){
        	System.out.println("ZkWebmccWatcher error: ");
        	e.printStackTrace();
        } catch (InterruptedException e){

        }
	}
}

